package org.example.util;

import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.example.modelTest.MqttTest;
import org.example.util.constantUtil.ConstantUtil;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
@Slf4j
public class MqttUtil {
    private final static int QOS_1 = 1;
    private final static String USER_NAME = "BS53CLIENT";
    private final static int PASSWORD = 13011301;
    private final static int KEEP_ALIVE = 60;
    //客户端唯一ID
    private static final String clientid = "11144";

    private boolean isMqttClientCreated = false;
    private MqttClient mqttClient = null;

    private final String TOPIC_PREFIX="zmh/";
    private final String PUB_TOPIC_SUFFIX = "/C2";
    private final String SUB_TOPIC_SUFFIX = "/C1";
    private final int TIMEOUT_MS = 5000;//最多等5秒

    public static void main(String[] args) throws Exception {
        MqttTest mqttTest=new MqttTest();
        String response=mqttTest.sendControlMessage("787986992669","test");
        System.out.println(response);
    }

    /**
     * 发送控制指令，消息以Hex形式收发
     * @param messageC2 要发送的控制指令消息
     * @param pub 发布的主题
     * @param sub 订阅的主题
     * @return 订阅的topic返回的消息
     * @throws Exception
     */
    public String sendControlMessage(String broker,String messageC2,String pub,String sub,String controlCode) throws Exception {
        String topicPub=pub;
        String topicSub=sub;
        System.out.println("topicPub:"+topicPub+" topicSub:"+topicSub);
        // 创建MqttClient
        if (!isMqttClientCreated) {
            init(broker);
            isMqttClientCreated = true;
        }
        // 现在的 转换为hex编码
        byte[] data = new byte[messageC2.length() / 2];
        for (int i = 0; i < data.length; i++) {
            int index = i * 2;
            int j = Integer.parseInt(messageC2.substring(index, index + 2), 16);
            data[i] = (byte) j;
        }
        MqttMessage mqttMessage = new MqttMessage(data);

        // 订阅响应主题并等待消息返回
        CountDownLatch latch = new CountDownLatch(1);
        final String[] response = {null};


        // 没有控制码，纯发送消息等待结果
        if(controlCode==null || controlCode.isEmpty()){
            mqttClient.subscribe(topicSub, (topic, message) -> {
                System.out.println("realTopic"+topic);

                if (topic.equals(topicSub)) {
                    // 现在的
                    System.out.println("msg received!");
                    StringBuilder MyData= new StringBuilder();
                    byte[] dataBytes = message.getPayload();
                    for (byte dataByte : dataBytes) {
                        MyData.append(String.format("%02x", dataByte));
                    }
                    response[0] = MyData.toString();
                    latch.countDown();
                }
            });
        } else {  // 存在控制码，需要额外判断控制码是否对应
            mqttClient.subscribe(topicSub, (topic, message) -> {
                System.out.println("realTopic"+topic);
                // 主题对应 && 控制码对应
                log.info("controlCode:{}",String.format("%02x", message.getPayload()[0]));
                if (topic.equals(topicSub) && controlCode.equals(String.format("%02x", message.getPayload()[0]))) {
                    // 消息符合要求，设置响应消息并countDown  原来的
//                response[0] = message.toString();
                    // 现在的
                    System.out.println("msg received! ");
                    StringBuilder MyData= new StringBuilder();
                    byte[] dataBytes = message.getPayload();
                    for (byte dataByte : dataBytes) {
                        MyData.append(String.format("%02x", dataByte));
                    }
                    log.debug("received data:{}", MyData.toString());
                    response[0] = MyData.toString();
                    latch.countDown();
                }
            });
        }
        // 发布控制消息，注意先订阅后发送
        try {
            mqttClient.publish(topicPub, mqttMessage);
//            System.out.println("MQTT客户端:" + clientid + "向服务器发送消息:{}" + mqttMessage);
            System.out.println("MQTT客户端:" + clientid + "向服务器发送消息:{}" + bytesToHex(data));
        } catch (MqttException e) {
            log.error("MQTT消息推送失败:",e);
//            e.printStackTrace();
            mqttClient.disconnect();
            mqttClient.close();
            return "message:"+"MQTT消息推送失败";
        }

        // 等待超时或者消息返回
//        String response = null;
        if (latch.await(TIMEOUT_MS, TimeUnit.MILLISECONDS)) {
            // 如果等待latch计数归零并且收到响应消息，返回响应消息
            mqttClient.disconnect();
            mqttClient.close();
            isMqttClientCreated = false;
            return response[0];
        } else {
            // 如果等待超时，返回null
            System.out.println("time out!");
            mqttClient.disconnect();
            mqttClient.close();
            isMqttClientCreated = false;
            return null;
        }
    }
    private void init(String broker) throws Exception {
        MqttConnectOptions connectOptions = new MqttConnectOptions();
        //断开之后自动重联
        connectOptions.setAutomaticReconnect(false);//true改为了flase @pulsar
        //设置会话心跳时间 代理和客户端在不发送消息的情况下可以忍受的最长时间
        connectOptions.setKeepAliveInterval(KEEP_ALIVE);
        //不建立持久会话
        connectOptions.setCleanSession(true);
        //用户名
        connectOptions.setUserName(USER_NAME);
        //密码
        connectOptions.setPassword(String.valueOf(PASSWORD).toCharArray());

        String clientId = MqttClient.generateClientId();
        try {
            mqttClient = new MqttClient(broker, clientid, new MemoryPersistence());
            //MQTT连接
            mqttClient.connect(connectOptions);
            //消息回调
//            client.setCallback(new MqttCallBackHandle(client));
        } catch (MqttException e) {
            log.error("MQTT消息异常:",e);
        }
//        mqttClient = new MqttClient(BROKER, clientId);
//        mqttClient.connect();
    }
    private static String bytesToHex(byte[] bytes) {
        StringBuilder sb = new StringBuilder();
        for (byte b : bytes) {
            sb.append(String.format("%02X", b));
        }
        return sb.toString();
    }
}
